package com.deep.flink;


import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Random;
import java.util.UUID;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;

/**
 * @author wangshida@baijia.com
 * @datetime 2022-03-20 下午12:03
 * @CopyRight (C) 百家互联
 * @desc 自定义数据源
 * @menu
 */
public class VideoOrderSource extends RichParallelSourceFunction<VideoOrder> {
    //把该变量声明为volatile（不稳定的），这就指示JVM，这个变量是不稳定的，每次使用它都到主存中进行读取,而不是从缓存读取
    private volatile Boolean flag = true;

    private Random random = new Random();

    private static List<String> list = new ArrayList<>();
    static {
        list.add("spring boot2.x");
        list.add("SpringCloud");
        list.add("RabbitMQ");
        list.add("Kafka");
        list.add("mianshi");
        list.add("FlinkStream");
        list.add("Linux");
    }

    @Override
    public void open(Configuration parameters) throws Exception {
//        super.open(parameters);
        System.out.println("自定义source，数据生产open---------");
    }

    @Override
    public void close() throws Exception {
//        super.close();
        System.out.println("自定义source，数据生产close---------");
    }

    @Override
    public void run(SourceContext<VideoOrder> ctx) throws Exception {
        while (flag){
            Thread.sleep(1000);
            String id = UUID.randomUUID().toString();
            int userId = random.nextInt(10);
            int money = random.nextInt(100);
            int videoNum = random.nextInt(list.size());
            String title = list.get(videoNum);
            VideoOrder videoOrder = new VideoOrder(id,title,money,userId,new Date());
            ctx.collect(videoOrder);
        }
    }

    /**
     * 结束时候执行
     */
    @Override
    public void cancel() {
        flag = false;
        System.out.println("关闭");
    }
}